package com.k8.exchange.impl;

import com.k8.common.executor.ExecutorRepository;
import com.k8.common.executor.NamedThreadFactory;
import com.k8.common.factroy.BeanFactory;
import com.k8.exchange.SerializeClientResolver;
import com.k8.exchange.EnvironmentUtil;
import com.k8.exchange.SerializeServerResolver;
import com.k8.exchange.ServerBuilder;
import com.k8.rpc.RpcException;
import com.k8.rpc.exchange.Exchange;
import com.k8.rpc.invoke.InvokeMode;
import com.k8.rpc.invoke.invocation.Invocation;
import com.k8.rpc.invoke.invocation.RpcInvocationTemplateManager;
import com.k8.rpc.result.Result;
import com.k8.rpc.result.ResultFuture;
import com.k8.rpc.result.RpcResult;
import com.k8.rpc.serialize.Serializer;
import com.k8.rpc.url.URL;
import com.k8.rpc.util.RpcUtils;
import com.key.w8.http1.connection.environment.ClientPoolEnvironment;
import com.key.w8.http1.connection.pool.ClientConnectionBuilder;
import com.key.w8.http1.connection.pool.ConnectionPool;
import com.key.w8.http1.connection.reconnect.ReConnectSupportConnectionBuilder;
import com.key.w8.http1.endpoint.HServer;
import com.key.w8.http1.entity.HttpObject;
import com.key.w8.http1.entity.HttpRequest;
import com.key.w8.http1.entity.HttpResponse;
import com.key.w8.http1.handler.codec.CodecBuilder;
import com.key.w8.http1.handler.codec.DefaultCodecBuilder;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpVersion;

import java.io.IOException;
import java.util.concurrent.*;

import static com.k8.common.constants.CommonKeyConstants.*;

/**
 * @Author: k8
 * @CreateTime: 2025-04-07
 * @Version: 1.0
 */
public class DefaultExchange implements Exchange {
    private static int DEFAULT_IO_TIME_OUT = 3000;
    protected int timeOut = DEFAULT_IO_TIME_OUT;
    private ConnectionPool pool;
    private Serializer serializer;
    private ExecutorService executorService;
    private String side;
    private volatile boolean started;
    private final Object startLock = new Object();
    private HServer hServer;

    @Override
    public boolean serverOpened() {
        return started;
    }

    public void openServer(URL protocolUrl, BeanFactory beanFactory) throws ExecutionException, InterruptedException {
        if (!PROVIDER.equals(protocolUrl.getSide()))
            throw new IllegalArgumentException("side of exchange is not provider");
        if (started) return;
        synchronized (startLock) {
            if (started) return;
            this.serializer = RpcUtils.getSerializer(protocolUrl);
            SerializeServerResolver serializeServerResolver = new SerializeServerResolver(serializer, beanFactory.getBean(RpcInvocationTemplateManager.class));
            this.hServer = new ServerBuilder()
                    .setResponseResolver(serializeServerResolver)
                    .setProtocolUrl(protocolUrl)
                    .setSerializer(serializer)
                    .build();
            //阻塞等待服务开启
            this.hServer.openServer().get();
            this.side = protocolUrl.getSide();
            started = true;
        }

    }

    public void preConnect(URL subUrl) {
        String side = subUrl.getSide();
        if (!CONSUMER.equals(side)) {
            throw new IllegalArgumentException("side of exchange is not consumer");
        }
        synchronized (startLock) {
            this.timeOut = subUrl.getParameter(IO_TIMEOUT_KEY, DEFAULT_IO_TIME_OUT);
            this.executorService = subUrl.getExtensionAccessor().getDefaultExtension(ExecutorRepository.class).getExecutorService(subUrl);
            this.serializer = RpcUtils.getSerializer(subUrl);
            this.side = side;
            ClientPoolEnvironment clientPoolEnvironment = EnvironmentUtil.urlConvert2ClientEnvironment(subUrl);
            ThreadFactory threadFactory = NamedThreadFactory.USER_THREAD_FACTORY;
            CodecBuilder codecBuilder = new DefaultCodecBuilder();
            codecBuilder.addHttpObjectClientResolver(new SerializeClientResolver(serializer));
            ClientConnectionBuilder connectionBuilder = new ReConnectSupportConnectionBuilder(clientPoolEnvironment, codecBuilder);
            this.pool = new ConnectionPool(clientPoolEnvironment.getMaxConnections(), clientPoolEnvironment.getMaxRequestSends(),
                    threadFactory, connectionBuilder);
        }

    }

    public Result<?> call(Invocation invocation) throws Throwable {
        HttpRequest request = new HttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/" + invocation.getInvocationKey());
        request.setComplexObject(invocation);
        Future<HttpResponse> httpResponseFuture = pool.writeRequestNeedResponse(request);
        InvokeMode invokeMode = invocation.getInvokeMode();

        switch (invokeMode) {
            //httpResponseFuture本身是支持异步的，但是需要进行包装，调用方法返回值类型毕竟不是HttpResponse
            case FUTURE -> {
                ResultFuture resultFuture = new ResultFuture(httpResponseFuture, (responseFuture, timeOut, timeUnit) -> {
                    try {
                        Future<HttpResponse> future = (Future<HttpResponse>) responseFuture;
                        HttpResponse response = null;
                        if (timeUnit == null || timeOut <= 0) {
                            response = future.get();
                        } else {
                            response = future.get(timeOut, timeUnit);
                        }
                        if (response.complexObject() == null || !(response.complexObject() instanceof Result<?>)) throw new IllegalStateException("返回出现异常，没有Result包装");
                        Result result = (Result) response.complexObject();
                        if (result.hasException()){
                            throw result.getException();
                        }
                        return result.getValue();
                    } catch (Throwable e) {
                        throw new RuntimeException(e);
                    }
                });
                Result<Object> result = new RpcResult<>();
                result.setValue(resultFuture);
                return result;
            }
            case ASYNC_FUTURE -> {
                Result<Object> result = new RpcResult<>();
                CompletableFuture<Object> future = CompletableFuture.supplyAsync(() -> {
                    try {
                        HttpResponse response = httpResponseFuture.get();
                        if (response == null) throw new RpcException("Call result error.");
                        Result<Object> result1 = (Result<Object>) response.complexObject();
                        if (result1.hasException()) {
                            throw result1.getException();
                        }
                        return result1.getValue();
                    } catch (Throwable e) {
                        throw new RuntimeException(e);
                    }
                }, executorService);
                result.setValue(future);
                return result;
            }
            default -> {
                Result<Object> result = getResult(httpResponseFuture);
                if (result.hasException()) {
                    throw result.getException();
                } else {
                    return result;
                }
            }
        }
    }

    @Override
    public void shutDown() {
        if (!started) return;
        synchronized (startLock) {
            if (!started) return;
            try {
                hServer.shutDown();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    private Result<Object> getResult(Future<HttpResponse> httpResponseFuture) throws ExecutionException, InterruptedException, TimeoutException {
        HttpResponse response;
        if (timeOut < 0) {
            response = httpResponseFuture.get();
        } else {
            response = httpResponseFuture.get(timeOut, TimeUnit.MILLISECONDS);
        }
        if (response == null) throw new RpcException("Call result error.");
        Object object = response.complexObject();
        if (object == null) throw new RpcException("Return type is not a Result.");
        return (Result<Object>) object;
    }


    //添加到codec，用于解析，给前缀

}
